package com.fudian.spark_platform.Clustering

import com.mongodb.spark.rdd.DocumentRDDFunctions
import com.fudian.spark_platform.MLClusteringConf
import org.apache.spark.sql.DataFrame
import org.bson.Document

case class fx168Row(words:String,url:String)

class FX168Clustering(conf: MLClusteringConf) {

    def clustering() = {

        val mLUtils = conf.mLUtils
        mLUtils.loadANSJDict()

        val spark = conf.spark
        val mongoData = conf.dataF.rdd
        val df = conf.dataF

        val waitRdd = df.select("html", "url").rdd.map(data => (data.get(0), data.get(1)))
        //通过对document的Rdd操作,取出每个文本,先进行分词操作,然后将得到的词组list和并,用空格链接各个元素后交给tokenize方法去实现过滤和特征转换
        val document = waitRdd.map(dataZ => {
            (mLUtils.ansjCut(dataZ._1.toString), dataZ._2)
        }).map(dataX => {
            (mLUtils.tokenize(dataX._1), dataX._2)
        }).map(dataF => {
            (mLUtils.isFanfa(dataF._1.asInstanceOf[Seq[String]]), dataF._2)
        }).filter(
            _._1.nonEmpty
        ).cache()
        //开始对多方面进行分析
        val documentCountTop10 = document.map(data => {
            (data._1.length, (data._1, data._2))
        }).sortBy(_._1, false).take(10).foreach(println(_))
        //然后是对单个过滤词的存在数量进行统计
        document.flatMap(_._1).map(x => {
            (x, 1)
        }).reduceByKey(_ + _).sortBy(_._2, false).take(10).foreach(println(_))
        //所有标记数量
        val totalPage = document.count()
        //所有的标记记录导出
        print("所有非法宣传页面的数量为: " + totalPage.toString)
        //这里必须将rdd转换成 rdd[Document]类型才能写入数据
        val documentD = document.map(x => {
            var tempStr = ""
            x._1.foreach(y => {
                tempStr += y + " "
            })
            (tempStr, x._2)
        }).map(i => {
            Document.parse("{\"words\":\"" + i._1 + "\",\"url\":\"" + i._2 + "\"}")
        })
        DocumentRDDFunctions.apply(documentD).saveToMongoDB()

        document.map(data => {
            fx168Row(data._1.asInstanceOf[String],data._2.asInstanceOf[String])
        })
    }

}
